package com.my.study.java.serialization;

import com.my.study.java.bean.Company;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

/**
 * @author Carlos
 * @version 1.0
 * @Description TODO
 * @date 2021/7/24 17:03
 **/
public class CompanyDeserializer implements Deserializer<Company> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public Company deserialize(String topic, byte[] data) {
        if (null == data) {
            return null;
        }
        if (data.length < 8) {
            throw new SerializationException("");
        }
        ByteBuffer buffer = ByteBuffer.wrap(data);
        int nameLen, addressLen;
        String name, address;

        nameLen = buffer.getInt();
        byte[] nameBytes = new byte[nameLen];
        buffer.get(nameBytes);

        addressLen = buffer.getInt();
        byte[] addressBytes = new byte[addressLen];
        buffer.get(addressBytes);

        try {
            name = new String(nameBytes, "UTF-8");
            address = new String(addressBytes, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("");
        }

        return new Company(name, address);
    }

    @Override
    public void close() {

    }

    public static void main(String[] args) {
        CompanySerializer companySerializer = new CompanySerializer();
        byte[] serialize = companySerializer.serialize("", new Company("ths", "同顺街道18号"));

        CompanyDeserializer companyDeserializer = new CompanyDeserializer();
        Company company = companyDeserializer.deserialize("", serialize);
        System.out.println(company);
    }


}
